/**
 * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 */

package com.iwindplus.gateway.server.listener;

import com.iwindplus.boot.domain.constant.CommonConstant.ExceptionConstant;
import com.iwindplus.boot.domain.constant.CommonConstant.NumberConstant;
import com.iwindplus.boot.domain.dto.ValidListDTO;
import com.iwindplus.gateway.server.domain.event.GatewayLogEvent;
import com.iwindplus.log.client.GatewayLogClient;
import com.iwindplus.log.domain.dto.GatewayLogDTO;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Lazy;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import reactor.core.scheduler.Schedulers;

/**
 * 网关日志监听器.
 *
 * @author zengdegui
 * @since 2025/03/21 21:51
 */
@Slf4j
@Component
public class GatewayLogListener {

    @Lazy
    @Resource
    private GatewayLogClient gatewayLogClient;

    private final BlockingQueue<GatewayLogDTO> logQueue = new LinkedBlockingQueue<>(1000);
    private CountDownLatch flushSignal = new CountDownLatch(1);

    @PostConstruct
    public void init() {
        Schedulers.boundedElastic().schedule(this::startBatchConsumer);
    }

    /**
     * 网关日志监听保存数据.
     *
     * @param gatewayLogEvent 网关日志事件
     */
    @Async
    @EventListener(GatewayLogEvent.class)
    public void onApplicationEvent(GatewayLogEvent gatewayLogEvent) {
        log.info("网关日志消息消费开始");
        final GatewayLogDTO logData = gatewayLogEvent.getLogData();

        if (logData != null) {
            logQueue.offer(logData);

            // 当队列中日志数量 >=10 条时触发提交
            if (logQueue.size() >= NumberConstant.NUMBER_TEN) {
                // 唤醒消费线程
                flushSignal.countDown();
            }
        }
    }


    private void startBatchConsumer() {
        List<GatewayLogDTO> batch = new ArrayList<>();
        while (!Thread.interrupted()) {
            try {
                // 等待直到有信号触发（即 countDown）
                flushSignal.await();

                int size = logQueue.drainTo(batch, NumberConstant.NUMBER_TEN);
                if (size > 0) {
                    processBatch(batch);
                    batch.clear();
                }

                // 重置 latch
                flushSignal = new CountDownLatch(1);
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();

                log.warn(ExceptionConstant.INTERRUPTED_EXCEPTION, ex);
                break;
            }
        }
    }

    private void processBatch(List<GatewayLogDTO> batch) {
        try {
            ValidListDTO<GatewayLogDTO> entity = new ValidListDTO<>(batch);
            this.gatewayLogClient.saveBatch(entity);

            log.info("成功提交网关日志，数量: {}", batch.size());
        } catch (Exception e) {
            log.error("批量保存网关日志失败", e);
        }
    }

}
